package io.github.wanggit.antrpc.console.service.impl;

import io.github.wanggit.antrpc.commons.bean.Host;
import io.github.wanggit.antrpc.commons.bean.IpNodeDataBean;
import io.github.wanggit.antrpc.commons.bean.RegisterBeanMethod;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.lb.LoadBalancerDTO;
import io.github.wanggit.antrpc.commons.utils.ApplicationNameUtil;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.zookeeper.NodeUnregDTO;
import io.github.wanggit.antrpc.console.domain.ConsumerRateLimiting;
import io.github.wanggit.antrpc.console.domain.ProviderRateLimiting;
import io.github.wanggit.antrpc.console.service.ConsumerRateLimitingService;
import io.github.wanggit.antrpc.console.service.InterfaceService;
import io.github.wanggit.antrpc.console.service.ProviderRateLimitingService;
import io.github.wanggit.antrpc.console.service.dto.InterfaceCalledDetailsDTO;
import io.github.wanggit.antrpc.console.service.utils.RegisterBeanMethodUtil;
import io.github.wanggit.antrpc.console.web.constants.WebConstants;
import io.github.wanggit.antrpc.console.web.utils.DoubleScaleUtil;
import io.github.wanggit.antrpc.console.web.utils.LongToDateUtil;
import io.github.wanggit.antrpc.console.web.vo.*;
import io.github.wanggit.antrpc.console.zookeeper.*;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Component
public class InterfaceServiceImpl implements InterfaceService {

    @Autowired private IZkNodeOperator zkNodeOperator;

    @Autowired private RestHighLevelClient restHighLevelClient;

    @Autowired private ISubscribeContainer subscribeContainer;

    @Autowired private IInterfaceContainer interfaceContainer;

    @Autowired private IUnregNodeContainer unregNodeContainer;

    @Autowired private ConsumerRateLimitingService consumerRateLimitingService;

    @Autowired private ProviderRateLimitingService providerRateLimitingService;

    @Value("${antrpc.default.load.balancer}")
    private String defaultLoadBalancer;

    @Override
    public void updateLoadbalancer(String className, String loadBalancer) {
        String fullPath = internalGetFullPath(className);
        byte[] nodeData = zkNodeOperator.getNodeData(fullPath);
        if (null == nodeData) {
            internalCreateNode(className, loadBalancer, fullPath);
        } else {
            LoadBalancerDTO loadBalancerDTO = convertByteArrayToLoadBalancerDTO(nodeData);
            if (null == loadBalancerDTO) {
                loadBalancerDTO = new LoadBalancerDTO();
            }
            loadBalancerDTO.setClazz(className);
            loadBalancerDTO.setLb(loadBalancer);
            zkNodeOperator.setNodeData(fullPath, JsonUtils.toJsonBytes(loadBalancerDTO));
        }
    }

    private LoadBalancerDTO convertByteArrayToLoadBalancerDTO(byte[] nodeData) {
        try {
            return JsonUtils.fromJson(nodeData, LoadBalancerDTO.class);
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error(
                        "parse json to "
                                + LoadBalancerDTO.class.getName()
                                + " error. "
                                + new String(nodeData, Charset.forName("UTF-8")));
            }
        }
        return null;
    }

    @Override
    public void updateWeight(String clazz, String appNameHostPort, int weight) {
        Host hostEntity = ApplicationNameUtil.getHost(appNameHostPort);
        String fullPath = internalGetFullPath(clazz);
        byte[] nodeData = zkNodeOperator.getNodeData(fullPath);
        if (null == nodeData) {
            LoadBalancerDTO loadBalancerDTO = new LoadBalancerDTO();
            loadBalancerDTO.setClazz(clazz);
            loadBalancerDTO.setLb(defaultLoadBalancer);
            Map<String, Integer> weights = new HashMap<>();
            weights.put(hostEntity.getHostInfo(), weight);
            loadBalancerDTO.setWeights(weights);
            zkNodeOperator.createNode(fullPath, JsonUtils.toJsonBytes(loadBalancerDTO));
        } else {
            LoadBalancerDTO loadBalancerDTO = convertByteArrayToLoadBalancerDTO(nodeData);
            if (null == loadBalancerDTO) {
                loadBalancerDTO = new LoadBalancerDTO();
            }
            Map<String, Integer> weights = loadBalancerDTO.getWeights();
            if (null == weights) {
                weights = new HashMap<>();
            }
            weights.put(hostEntity.getHostInfo(), weight);
            loadBalancerDTO.setWeights(weights);
            loadBalancerDTO.setClazz(clazz);
            zkNodeOperator.setNodeData(fullPath, JsonUtils.toJsonBytes(loadBalancerDTO));
        }
    }

    @Override
    public void classOnline(String appNameHostPort, String clazz) {
        Host hostEntity = ApplicationNameUtil.getHost(appNameHostPort);
        String hostInfo = hostEntity.getHostInfo();
        String fullPath = "/" + ConstantValues.ZK_ROOT_NODE_NAME + "/" + hostInfo;
        boolean existsNode = zkNodeOperator.existsNode(fullPath);
        if (!existsNode) {
            IpNodeDataBean ipNodeDataBean = new IpNodeDataBean();
            ipNodeDataBean.setAppName(appNameHostPort);
            ipNodeDataBean.setRpcPort(hostEntity.getPort());
            ipNodeDataBean.setTs(System.currentTimeMillis());
            zkNodeOperator.createNode(fullPath, JsonUtils.toJsonBytes(ipNodeDataBean));
        }
        zkNodeOperator.deleteNode(createOfflinePath(hostInfo, clazz));
    }

    @Override
    public void classOffline(String appNameHostPort, String clazz) {
        String hostAndPort = ApplicationNameUtil.getHost(appNameHostPort).getHostInfo();
        NodeUnregDTO nodeUnregDTO = new NodeUnregDTO();
        nodeUnregDTO.setTs(System.currentTimeMillis());
        nodeUnregDTO.setHost(hostAndPort);
        nodeUnregDTO.setAppName(ApplicationNameUtil.getSimpleAppName(appNameHostPort));
        nodeUnregDTO.setClazz(clazz);
        zkNodeOperator.createNode(
                createOfflinePath(hostAndPort, clazz), JsonUtils.toJsonBytes(nodeUnregDTO));
    }

    @Override
    public InterfaceDetails collectInterfaceDetails(String className) {
        InterfaceDetails interfaceDetails = new InterfaceDetails();
        interfaceDetails.setClassName(className);
        // 1.查询这个接口的负载均衡器，权重、负载均衡
        LoadBalancerDTO loadBalancerDTO = queryLoadBalancer(className);
        if (null == loadBalancerDTO) {
            loadBalancerDTO = new LoadBalancerDTO();
            loadBalancerDTO.setClazz(className);
            loadBalancerDTO.setLb(defaultLoadBalancer);
            loadBalancerDTO.setWeights(new HashMap<>());
        }
        interfaceDetails.setClassName(loadBalancerDTO.getClazz());
        interfaceDetails.setLoadBalancerName(
                null == loadBalancerDTO.getLb() ? defaultLoadBalancer : loadBalancerDTO.getLb());
        // 2.查询这个接口的所有在线的提供者，注册时间、AppName
        List<InterfaceDTO> interfaceDTOList =
                interfaceContainer.findContainsByInterfaceName(className);
        // 3.查询这个接口已被下线的提供者，下线时间
        List<NodeUnregDTO> nodeUnregDTOs = unregNodeContainer.findNodeUnregDTOs(className);
        // 4.查询这个接口的所有引用者， 订阅时间、订阅AppName
        List<SubscribeNodeVO> subscribeNodeVOList = subscribeContainer.findByClassName(className);
        List<InterfaceConsumerDetail> consumerDetails = new ArrayList<>();
        if (null != subscribeNodeVOList) {
            for (SubscribeNodeVO subscribeNodeVO : subscribeNodeVOList) {
                InterfaceConsumerDetail consumerDetail = new InterfaceConsumerDetail();
                consumerDetail.setAppName(subscribeNodeVO.getHost());
                consumerDetail.setRegTime(subscribeNodeVO.getTsStr());
                consumerDetails.add(consumerDetail);
            }
        }
        interfaceDetails.setConsumers(consumerDetails);
        // 5.查询这个接口各个服务提供者的调用日志
        List<InterfaceProviderDetail> providerDetails = new ArrayList<>();
        for (InterfaceDTO interfaceDTO : interfaceDTOList) {
            InterfaceProviderDetail providerDetail = new InterfaceProviderDetail();
            providerDetail.setAppName(interfaceDTO.getAppName() + "@" + interfaceDTO.getHost());
            providerDetail.setRegTime(LongToDateUtil.toDate(interfaceDTO.getRegisterTs()));
            Map<String, Integer> weights = loadBalancerDTO.getWeights();
            if (null == weights) {
                providerDetail.setWeight(1);
            } else {
                providerDetail.setWeight(weights.getOrDefault(interfaceDTO.getHost(), 1));
            }
            providerDetails.add(providerDetail);
        }
        for (NodeUnregDTO nodeUnregDTO : nodeUnregDTOs) {
            InterfaceProviderDetail providerDetail = new InterfaceProviderDetail();
            providerDetail.setAppName(nodeUnregDTO.getAppName() + "@" + nodeUnregDTO.getHost());
            providerDetail.setUnregTime(LongToDateUtil.toDate(nodeUnregDTO.getTs()));
            providerDetail.setWeight(0);
            providerDetails.add(providerDetail);
        }
        interfaceDetails.setProviders(providerDetails);
        return interfaceDetails;
    }

    @Override
    public LoadBalancerDTO queryLoadBalancer(String className) {
        byte[] nodeData =
                zkNodeOperator.getNodeData(
                        "/" + ConstantValues.ZK_ROOT_LB_NODE_NAME + "/" + className);
        if (null == nodeData) {
            return null;
        }
        return JsonUtils.fromJson(nodeData, LoadBalancerDTO.class);
    }

    @Override
    public List<InterfaceCalledDetailsDTO> queryInterfaceCalledDetails(
            String className, String appNameHostPort, String start, String end) {
        Host hostEntity = ApplicationNameUtil.getHost(appNameHostPort);
        SearchSourceBuilder searchSourceBuilder =
                createSearchSourceBuilder(className, start, end, hostEntity);
        SearchRequest searchRequest = new SearchRequest(WebConstants.ELASTIC_CALL_LOGS_IDX);
        searchRequest.source(searchSourceBuilder);
        List<InterfaceCalledDetailsDTO> detailsDTOS = new ArrayList<>();
        try {
            SearchResponse searchResponse =
                    restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = searchResponse.getAggregations();
            Terms methodNameAggregation = aggregations.get("targetMethodName");
            List<? extends Terms.Bucket> buckets = methodNameAggregation.getBuckets();
            if (null != buckets && !buckets.isEmpty()) {
                buckets.forEach(
                        it ->
                                detailsDTOS.add(
                                        fullDataToInterfaceCalledDetailsDTO(
                                                className, hostEntity, it)));
            }
        } catch (ElasticsearchStatusException ese) {
            log.error(ese.getMessage(), ese);
        } catch (IOException e) {
            if (log.isErrorEnabled()) {
                log.error("query call details error.", e);
            }
        }
        detailsDTOS.sort((o1, o2) -> o1.getMethodName().compareToIgnoreCase(o2.getMethodName()));
        return detailsDTOS;
    }

    @Override
    public void deleteZkNode(String path) {
        zkNodeOperator.deleteNode(path);
    }

    /**
     * 更新服务提供都频控
     *
     * @param className class name
     * @param humanFullMethodName human full method name
     * @param limit limit
     * @param durationInSeconds duration in seconds
     * @param appNameHostPort app host info
     */
    @Override
    public void serviceProviderUpdateRateLimit(
            String className,
            String humanFullMethodName,
            Integer limit,
            Integer durationInSeconds,
            String appNameHostPort) {
        Host host = ApplicationNameUtil.getHost(appNameHostPort);
        String ipAndPort = host.getHostInfo();
        InterfaceDTO interfaceDTO =
                interfaceContainer.findByInterfaceNameAndHostInfo(className, ipAndPort);
        if (null == interfaceDTO) {
            if (log.isErrorEnabled()) {
                log.error("Not Found service provider " + className + "@" + appNameHostPort);
            }
            return;
        }
        RegisterBeanMethod registerBeanMethod =
                findRegisterBeanMethodFromInterfaceDTO(humanFullMethodName, interfaceDTO);
        if (null == registerBeanMethod) {
            if (log.isErrorEnabled()) {
                log.error(
                        "Not Found method "
                                + humanFullMethodName
                                + " in service provider "
                                + className
                                + "@"
                                + appNameHostPort);
            }
            return;
        }
        registerBeanMethod.setLimit(limit);
        registerBeanMethod.setDurationInSeconds(durationInSeconds);
        zkNodeOperator.setNodeData(
                interfaceDTO.getPath(),
                JsonUtils.toJsonBytes(interfaceDTO.toInterfaceNodeDataBean()));
        // 保存限流信息到数据，以便于在节点重新注册时恢复
        saveProviderRateLimitingToDB(className, ipAndPort, registerBeanMethod);
    }

    /**
     * 更新服务订阅者端的频控
     *
     * @param className class name
     * @param humanFullMethodName human full method name
     * @param limit limit
     * @param durationInSeconds duration in seconds
     * @param appNameHostPort app host info
     */
    @Override
    public void serviceConsumerUpdateRateLimit(
            String className,
            String humanFullMethodName,
            Integer limit,
            Integer durationInSeconds,
            String appNameHostPort) {
        SubscribeNodeVO subscribeNodeVO =
                subscribeContainer.findByClassNameAndHost(className, appNameHostPort);
        if (null == subscribeNodeVO) {
            if (log.isErrorEnabled()) {
                log.error("Not Found service consumer " + className + "@" + appNameHostPort);
            }
            return;
        }
        Map<String, RegisterBeanMethod> methodMap =
                null == subscribeNodeVO.getMethodMap()
                        ? new HashMap<>()
                        : subscribeNodeVO.getMethodMap();
        List<InterfaceDTO> interfaceDTOS = interfaceContainer.findByInterfaceName(className);
        if (null == interfaceDTOS || interfaceDTOS.isEmpty()) {
            if (log.isErrorEnabled()) {
                log.error("Not Found service provider " + className);
            }
            return;
        }
        RegisterBeanMethod registerBeanMethod =
                findRegisterBeanMethodFromInterfaceDTO(humanFullMethodName, interfaceDTOS.get(0));
        if (null == registerBeanMethod) {
            if (log.isErrorEnabled()) {
                log.error(
                        "Not Found method "
                                + humanFullMethodName
                                + " in service provider "
                                + className
                                + "@"
                                + appNameHostPort);
            }
            return;
        }
        RegisterBeanMethod consumerRegisterBeanMethod =
                RegisterBeanMethodUtil.copy(registerBeanMethod);
        consumerRegisterBeanMethod.setLimit(limit);
        consumerRegisterBeanMethod.setDurationInSeconds(durationInSeconds);
        consumerRegisterBeanMethod.generateFullName();
        methodMap.put(consumerRegisterBeanMethod.getHumanFullName(), consumerRegisterBeanMethod);
        subscribeNodeVO.setMethodMap(methodMap);
        zkNodeOperator.setNodeData(
                "/"
                        + ConstantValues.ZK_ROOT_SUBSCRIBE_NODE_NAME
                        + "/"
                        + appNameHostPort
                        + "/"
                        + className,
                JsonUtils.toJsonBytes(subscribeNodeVO.toSubscribeNode()));
        // 保存到数据库，便于下次恢复
        saveSubscribeRateLimitingToDB(subscribeNodeVO, consumerRegisterBeanMethod);
    }

    @Override
    public List<MethodLimitVO> findConsumersByInterfaceName(
            String className, String humanFullMethodName) {
        List<SubscribeNodeVO> subscribeNodeVOS = subscribeContainer.findByClassName(className);
        return subscribeNodeVOS.stream()
                .map(
                        it -> {
                            MethodLimitVO methodLimitVO = new MethodLimitVO();
                            methodLimitVO.setAppNameAndHostInfo(it.getHost());
                            if (null != it.getMethodMap()) {
                                RegisterBeanMethod registerBeanMethod =
                                        it.getMethodMap().get(humanFullMethodName);
                                if (null != registerBeanMethod) {
                                    methodLimitVO.setDurationInSeconds(
                                            registerBeanMethod.getDurationInSeconds());
                                    methodLimitVO.setLimit(registerBeanMethod.getLimit());
                                }
                            }
                            return methodLimitVO;
                        })
                .collect(Collectors.toList());
    }

    private RegisterBeanMethod findRegisterBeanMethodFromInterfaceDTO(
            String humanFullMethodName, InterfaceDTO interfaceDTO) {
        return interfaceDTO.getInterfaceMethodDTOS().stream()
                .filter(
                        it -> {
                            it.generateFullName();
                            return Objects.equals(it.getHumanFullName(), humanFullMethodName);
                        })
                .findFirst()
                .orElse(null);
    }

    private void saveProviderRateLimitingToDB(
            String className, String ipAndPort, RegisterBeanMethod registerBeanMethod) {
        ProviderRateLimiting providerRateLimiting = new ProviderRateLimiting();
        providerRateLimiting.setParameterTypeNames(
                JsonUtils.toJsonString(registerBeanMethod.getParameterTypeNames()));
        providerRateLimiting.setMethodName(registerBeanMethod.getMethodName());
        providerRateLimiting.setMethodMd5(
                ProviderRateLimiting.generateMethodMd5(
                        ipAndPort,
                        className,
                        registerBeanMethod.getMethodName(),
                        JsonUtils.toJsonString(registerBeanMethod.getParameterTypeNames())));
        providerRateLimiting.setClassMd5(
                ProviderRateLimiting.generateClassMd5(ipAndPort, className));
        providerRateLimiting.setLimitValue(registerBeanMethod.getLimit());
        providerRateLimiting.setHostPort(ipAndPort);
        providerRateLimiting.setDurationInSeconds(registerBeanMethod.getDurationInSeconds());
        providerRateLimiting.setClassName(className);
        providerRateLimitingService.saveOrUpdate(providerRateLimiting);
    }

    private void saveSubscribeRateLimitingToDB(
            SubscribeNodeVO subscribeNodeVO, RegisterBeanMethod consumerRegisterBeanMethod) {
        ConsumerRateLimiting consumerRateLimiting = new ConsumerRateLimiting();
        consumerRateLimiting.setParameterTypeNames(
                JsonUtils.toJsonString(consumerRegisterBeanMethod.getParameterTypeNames()));
        consumerRateLimiting.setMethodName(consumerRegisterBeanMethod.getMethodName());
        consumerRateLimiting.setClassMd5(
                ConsumerRateLimiting.generateClassMd5(
                        subscribeNodeVO.getHost(), subscribeNodeVO.getClassName()));
        consumerRateLimiting.setMethodMd5(
                ConsumerRateLimiting.generateMethodMd5(
                        subscribeNodeVO.getHost(),
                        subscribeNodeVO.getClassName(),
                        consumerRegisterBeanMethod.getMethodName(),
                        JsonUtils.toJsonString(
                                consumerRegisterBeanMethod.getParameterTypeNames())));
        consumerRateLimiting.setLimitValue(consumerRegisterBeanMethod.getLimit());
        consumerRateLimiting.setDurationInSeconds(
                consumerRegisterBeanMethod.getDurationInSeconds());
        consumerRateLimiting.setAppNameHostPort(subscribeNodeVO.getHost());
        consumerRateLimiting.setClassName(subscribeNodeVO.getClassName());
        consumerRateLimitingService.saveOrUpdate(consumerRateLimiting);
    }

    private InterfaceCalledDetailsDTO fullDataToInterfaceCalledDetailsDTO(
            String className, Host hostEntity, Terms.Bucket bucket) {
        Aggregations bucketAggregations = bucket.getAggregations();
        ValueCount valueCount = bucketAggregations.get("count");
        long count = valueCount.getValue();
        Avg avg = bucketAggregations.get("avgRt");
        double avgValue = avg.getValue();
        Max max = bucketAggregations.get("maxRt");
        double maxValue = max.getValue();
        Min min = bucketAggregations.get("minRt");
        double minValue = min.getValue();
        Map<String, Double> percentRtDatas = new HashMap<>();
        parsePercentDatas(bucketAggregations, percentRtDatas);
        double rt1 = percentRtDatas.getOrDefault("rt1", 0.0);
        double rt5 = percentRtDatas.getOrDefault("rt5", 0.0);
        double rt25 = percentRtDatas.getOrDefault("rt25", 0.0);
        double rt50 = percentRtDatas.getOrDefault("rt50", 0.0);
        double rt75 = percentRtDatas.getOrDefault("rt75", 0.0);
        double rt95 = percentRtDatas.getOrDefault("rt95", 0.0);
        double rt99 = percentRtDatas.getOrDefault("rt99", 0.0);
        long qps1 = statInterfaceQps(className, hostEntity, WebConstants.TimeRanges.m1);
        long qps5 = statInterfaceQps(className, hostEntity, WebConstants.TimeRanges.m5);
        long qps15 = statInterfaceQps(className, hostEntity, WebConstants.TimeRanges.m15);
        InterfaceCalledDetailsDTO calledDetailsDTO = new InterfaceCalledDetailsDTO();
        calledDetailsDTO.setMethodName(bucket.getKeyAsString());
        calledDetailsDTO.setCount(count);
        calledDetailsDTO.setQps1(qps1);
        calledDetailsDTO.setQps5(qps5);
        calledDetailsDTO.setQps15(qps15);
        calledDetailsDTO.setRt1(DoubleScaleUtil.scale(rt1, 2));
        calledDetailsDTO.setRt5(DoubleScaleUtil.scale(rt5, 2));
        calledDetailsDTO.setRt25(DoubleScaleUtil.scale(rt25, 2));
        calledDetailsDTO.setRt50(DoubleScaleUtil.scale(rt50, 2));
        calledDetailsDTO.setRt75(DoubleScaleUtil.scale(rt75, 2));
        calledDetailsDTO.setRt95(DoubleScaleUtil.scale(rt95, 2));
        calledDetailsDTO.setRt99(DoubleScaleUtil.scale(rt99, 2));
        calledDetailsDTO.setRtMax(DoubleScaleUtil.scale(maxValue, 2));
        calledDetailsDTO.setRtMean(DoubleScaleUtil.scale(avgValue, 2));
        calledDetailsDTO.setRtMin(DoubleScaleUtil.scale(minValue, 2));
        return calledDetailsDTO;
    }

    private SearchSourceBuilder createSearchSourceBuilder(
            String className, String start, String end, Host hostEntity) {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        createBasicQuery(className, hostEntity, start, end, boolQueryBuilder);
        searchSourceBuilder.query(boolQueryBuilder);
        TermsAggregationBuilder termsAggregationBuilder =
                AggregationBuilders.terms("targetMethodName").field("targetMethodName.keyword");
        termsAggregationBuilder.subAggregation(
                AggregationBuilders.count("count").field("requestId.keyword"));
        termsAggregationBuilder.subAggregation(AggregationBuilders.max("maxRt").field("rt"));
        termsAggregationBuilder.subAggregation(AggregationBuilders.min("minRt").field("rt"));
        termsAggregationBuilder.subAggregation(AggregationBuilders.avg("avgRt").field("rt"));
        termsAggregationBuilder.subAggregation(AggregationBuilders.percentiles("pRt").field("rt"));
        searchSourceBuilder.aggregation(termsAggregationBuilder);
        searchSourceBuilder.fetchSource(false);
        return searchSourceBuilder;
    }

    private void parsePercentDatas(
            Aggregations bucketAggregations, Map<String, Double> percentRtDatas) {
        ParsedPercentiles parsedPercentiles = bucketAggregations.get("pRt");
        for (Percentile percentile : parsedPercentiles) {
            WebConstants.PercentRt percentRt =
                    WebConstants.PercentRt.instance(percentile.getPercent());
            if (null != percentRt) {
                percentRtDatas.put(percentRt.name(), percentile.getValue());
            }
        }
    }

    private long statInterfaceQps(
            String className, Host hostEntity, WebConstants.TimeRanges timeRanges) {
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        DateTime standard = new DateTime();
        String end = standard.toDate().toInstant().toString();
        String start =
                standard.minusMinutes(timeRanges.getMinutes()).toDate().toInstant().toString();
        createBasicQuery(className, hostEntity, start, end, boolQueryBuilder);
        TermsAggregationBuilder termsAggregationBuilder =
                AggregationBuilders.terms("targetMethodName").field("targetMethodName.keyword");
        termsAggregationBuilder.subAggregation(
                AggregationBuilders.count("count").field("requestId.keyword"));
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(boolQueryBuilder);
        searchSourceBuilder.aggregation(termsAggregationBuilder);
        searchSourceBuilder.fetchSource(false);
        SearchRequest searchRequest = new SearchRequest(WebConstants.ELASTIC_CALL_LOGS_IDX);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse =
                    restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = searchResponse.getAggregations();
            Terms methodNameAggregation = aggregations.get("targetMethodName");
            List<? extends Terms.Bucket> buckets = methodNameAggregation.getBuckets();
            if (!buckets.isEmpty()) {
                Terms.Bucket bucket = buckets.get(0);
                Aggregations bucketAggregations = bucket.getAggregations();
                ValueCount valueCount = bucketAggregations.get("count");
                return valueCount.getValue();
            }
        } catch (ElasticsearchStatusException ese) {
            log.error(ese.getMessage(), ese);
        } catch (IOException e) {
            if (log.isErrorEnabled()) {
                log.error("query qps " + timeRanges.name() + " error. ", e);
            }
        }
        return 0;
    }

    private void createBasicQuery(
            String className,
            Host hostEntity,
            String start,
            String end,
            BoolQueryBuilder boolQueryBuilder) {
        MatchQueryBuilder classNameMatchQueryBuilder =
                QueryBuilders.matchQuery("targetClassName", className);
        boolQueryBuilder.must(classNameMatchQueryBuilder);
        MatchQueryBuilder ipMatchQueryBuilder =
                QueryBuilders.matchQuery("targetIp", hostEntity.getIp());
        boolQueryBuilder.must(ipMatchQueryBuilder);
        MatchQueryBuilder portMatchQueryBuilder =
                QueryBuilders.matchQuery("targetPort", hostEntity.getPort());
        boolQueryBuilder.must(portMatchQueryBuilder);
        RangeQueryBuilder dateRangeQueryBuilder =
                QueryBuilders.rangeQuery("date").gte(start).lt(end);
        boolQueryBuilder.must(dateRangeQueryBuilder);
    }

    private String createOfflinePath(String host, String clazz) {
        return "/" + ConstantValues.ZK_ROOT_NODE_UNREG_NAME + "/" + host + "/" + clazz;
    }

    private void internalCreateNode(String className, String loadBalancer, String fullPath) {
        LoadBalancerDTO loadBalancerDTO = new LoadBalancerDTO();
        loadBalancerDTO.setLb(loadBalancer);
        loadBalancerDTO.setClazz(className);
        loadBalancerDTO.setWeights(new HashMap<>());
        zkNodeOperator.createNode(fullPath, JsonUtils.toJsonBytes(loadBalancerDTO));
    }

    private String internalGetFullPath(String className) {
        return "/" + ConstantValues.ZK_ROOT_LB_NODE_NAME + "/" + className;
    }
}
